package openase.audiosearch.server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import openase.audio.H2Storage;
import openase.audio.Storage;
import openase.audiosearch.job.Job;
import openase.audiosearch.job.factory.JobFactory;

import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class AudioSearchServer {

	private static final Log logger = LogFactory.getLog(AudioSearchServer.class);

	static {
		// configures the H2 storage to be used.
		Storage.setStorage(new H2Storage());
	}

	private ServerSocketChannel serverChannel;
	private Selector selector;
	private ExecutorService service = Executors.newCachedThreadPool();

	private volatile boolean upAndRunning = false;

	public AudioSearchServer() {
	}

	/**
	 * @throws IllegalStateException
	 */
	public synchronized void start() {
		if (upAndRunning) {
			throw new IllegalStateException("Server already running");
		}

		logger.info("Starting Open-ASE server...");
		try {
			serverChannel = createServer();
		} catch (IOException e) {
			throw new IllegalStateException("IO error during server creation", e);
		}
		logger.info("Server started. Listening port " + serverChannel.socket().getLocalPort());
		try {
			selector = Selector.open();
		} catch (IOException e) {
			throw new IllegalStateException("IO error during selector creation", e);
		}

		new Thread(new RequestReceiver(), "OpenASEServer-RequestReceiver").start();

		upAndRunning = true;
	}

	public boolean isRunning() {
		return upAndRunning;
	}

	/**
	 * @throws IllegalStateException
	 */
	public synchronized void stop() {
		if (!upAndRunning) {
			throw new IllegalStateException("Server isn't running");
		}

		upAndRunning = false;
		selector.wakeup();
		service.shutdown();

		try {
			if (!service.awaitTermination(1, TimeUnit.MINUTES)) {
				logger.warn("Jobs didn't finished in time.");
			}
		} catch (InterruptedException e) {
			logger.warn("Jobs didn't finished in time.", e);
		}

		try {
			serverChannel.close();
		} catch (IOException e) {
			logger.error("Error closing server.", e);
		}
		serverChannel = null;
		try {
			selector.close();
		} catch (IOException e) {
			logger.error("Error server's selector.", e);
		}
		selector = null;
	}

	private static ServerSocketChannel createServer() throws IOException {
		ServerSocketChannel serverChannel = ServerSocketChannel.open();
		serverChannel.socket().bind(new InetSocketAddress(AudioSearchServerProperties.SERVER_PORT));
		serverChannel.configureBlocking(false);

		return serverChannel;
	}

	private class RequestReceiver implements Runnable {

		public void run() {
			SelectionKey acceptKey;
			try {
				acceptKey = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
			} catch (ClosedChannelException e) {
				logger.error("The server was closed! ", e);
				return;
			}

			while (upAndRunning) {
				int selected;
				try {
					selected = selector.select(1000);
				} catch (IOException e) {
					logger.error("The selector failed! ", e);
					break;
				}
				if (selected > 0) {
					Iterator<SelectionKey> it = selector.selectedKeys().iterator();

					while (it.hasNext()) {
						SelectionKey sk = it.next();
						it.remove();

						final Job job = JobFactory.getJob(sk);
						if (job == null) {
							logger.warn(String.format("No job registered for key [%s : OPS = %d].", sk, sk.interestOps()));
							// sk.cancel();
							IOUtils.closeQuietly(sk.channel());
						} else {
							service.execute(job);
							logger.info(String.format("Job scheduled to execute: %s", job));
						}
						if (sk != acceptKey) {
							sk.cancel();
						}
					}
				}

			}
		}

	}

}
